package app.function;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import app.bean.CategoryMiningResult;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

/**
 * 品类挖掘处理函数 - 实现实时挖掘分析
 */
public class CategoryMiningProcessFunction extends KeyedProcessFunction<String, CategoryMiningResult, CategoryMiningResult> {
    
    private final String miningType;
    private final int topN;
    private ListState<CategoryMiningResult> categoryListState;
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public CategoryMiningProcessFunction(String miningType, int topN) {
        this.miningType = miningType;
        this.topN = topN;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        categoryListState = getRuntimeContext().getListState(
            new ListStateDescriptor<>("categoryListState", 
                TypeInformation.of(new TypeHint<CategoryMiningResult>() {}))
        );
    }
    
    @Override
    public void processElement(CategoryMiningResult value, Context ctx, Collector<CategoryMiningResult> out) throws Exception {
        // 添加到状态列表
        categoryListState.add(value);
        
        // 注册定时器，延迟1秒处理挖掘
        ctx.timerService().registerEventTimeTimer(ctx.timerService().currentWatermark() + 1000);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<CategoryMiningResult> out) throws Exception {
        // 获取所有品类数据
        List<CategoryMiningResult> categoryList = new ArrayList<>();
        for (CategoryMiningResult category : categoryListState.get()) {
            categoryList.add(category);
        }
        
        // 计算总销售额用于市场份额计算
        double totalMarketSales = categoryList.stream()
                .mapToDouble(c -> c.getTotalSalesAmount() != null ? c.getTotalSalesAmount() : 0.0)
                .sum();
        
        // 为每个品类计算挖掘指标
        for (CategoryMiningResult category : categoryList) {
            // 计算市场份额
            if (totalMarketSales > 0 && category.getTotalSalesAmount() != null) {
                category.setMarketShare(category.getTotalSalesAmount() / totalMarketSales * 100);
            }
            
            // 设置当前和上期销售数据（模拟）
            category.setCurrentSales(category.getTotalSalesAmount());
            category.setPreviousSales(category.getTotalSalesAmount() * 0.8); // 模拟上期数据
            
            // 计算增长率
            category.calculateGrowthRate();
            
            // 设置需求和供给数据（模拟）
            category.setDemandCount(category.getTotalSalesCount() * 2); // 模拟需求量
            category.setSupplyCount(category.getProductCount());
            
            // 计算供需比
            category.calculateSupplyDemandRatio();
            
            // 分析品类类型
            category.analyzeCategoryType();
            
            // 计算机会分数
            category.calculateOpportunityScore();
            
            // 设置实时数仓标准字段
            category.setCurrentTimestamp();
            category.setDataLineage(DATA_SOURCE, VERSION);
            category.setPeriod("7d"); // 7天统计周期
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long windowStart = timestamp - 604800000; // 7天前
            category.setWindowTime(sdf.format(new Date(windowStart)), sdf.format(new Date(timestamp)));
        }
        
        // 根据挖掘类型进行排序
        switch (miningType) {
            case "opportunity":
                categoryList.sort(Comparator.comparing(CategoryMiningResult::getOpportunityScore, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "growth":
                categoryList.sort(Comparator.comparing(CategoryMiningResult::getGrowthRate, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            case "market_share":
                categoryList.sort(Comparator.comparing(CategoryMiningResult::getMarketShare, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
                break;
            default:
                categoryList.sort(Comparator.comparing(CategoryMiningResult::getTotalSalesAmount, Comparator.nullsLast(Comparator.naturalOrder())).reversed());
        }
        
        // 输出前TopN品类，设置排名
        for (int i = 0; i < Math.min(topN, categoryList.size()); i++) {
            CategoryMiningResult category = categoryList.get(i);
            category.setRank(i + 1);
            category.setRankType(miningType);
            
            out.collect(category);
        }
        
        // 清空状态
        categoryListState.clear();
    }
}